/*实现异步工作器*/

#include "buffer.hpp"
#include <thread>
#include <condition_variable>
#include <atomic>
#include <mutex>
#include <functional>
namespace lcllog
{   
    enum class AsyncType
    {
        SAFE, // 安全状态，即缓冲区满了阻塞，避免资源耗尽
        UNSAFE // 不考虑资源耗尽问题，无限扩容，用来测试性能
    };
    class AsynWorker
    {
    public:
        using Functor = std::function<void(Buffer& buffer)>;
        using ptr = std::shared_ptr<AsynWorker>;
        AsynWorker(const Functor& callback, AsyncType worker_type = AsyncType::SAFE)
        :_callBack(callback)
        ,_worker_type(worker_type)
        ,_running(true)
        ,_thread(std::thread(&AsynWorker::threadEntry, this))
        {}
        ~AsynWorker() { stop(); }
        void stop()
        {
            _running = false; // 将运行状态设为false
            _cond_pop.notify_all(); 
            _thread.join();
        }
        void push(const char* data, size_t len)
        {
            // 1. 无限扩容(测试性能)  2. 固定大小 -- 当生产缓冲区满了时就阻塞
            if(_running == false) return;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                if(_worker_type == AsyncType::SAFE)
                    //条件变量控制，若生产缓冲区剩余空间大小大于等于数据长度，则可以添加数据，安全做法
                    _cond_push.wait(lock, [&](){ return _task_push.writeAbleSize() >= len; });
                _task_push.push(data, len);
            }
            // 唤醒消费者对数据进行处理
            _cond_pop.notify_all();
        }
    private:
        void threadEntry() // 线程入口函数
        {
            while(true)
            {
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    // 工作器没有运行，并且任务队列中的数据为空时，不用处理
                    if (_running == false && _task_push.empty()) { return; }
                    // 判断生产缓冲区有没有数据，有就处理，没有阻塞
                    // 或者异步工作器退出时被唤醒，检查有没有数据进行处理
                    _cond_pop.wait(lock, [&]{ return !_task_push.empty() || !_running; });
                    _task_push.swap(_task_pop);
                }
                if(_worker_type == AsyncType::SAFE)
                    // 唤醒生产者，不是安全状态不需要，因为就没有被阻塞
                    _cond_push.notify_all();
                // 对消费缓冲区的数据进行处理
                _callBack(_task_pop);
                // 初始化消费缓冲区
                _task_pop.reset();
            }
        }
    private:
        AsyncType _worker_type;
        std::atomic<bool> _running; // 异步工作器是否启动
        Buffer _task_push; // 生产缓冲区
        Buffer _task_pop;  // 消费缓冲区
        std::condition_variable _cond_push;
        std::condition_variable _cond_pop;
        std::mutex _mutex;
        std::thread _thread; // 异步工作器的工作线程
        Functor _callBack; // 具体对缓冲区数据进行处理的回调函数，由异步工作器使用者传入
    };
}